package com.celeritymq.handler;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

import javax.jms.Destination;
import javax.jms.JMSException;

import com.celeritymq.cache.Channel;
import com.celeritymq.cache.ChannelFactory;
import com.celeritymq.destination.DestinationContainer;
import com.celeritymq.message.CelerityMessage;
import com.celeritymq.utils.CelerityBuffer;
import com.celeritymq.utils.ObjectUtil;

public class CelerityReaderImpl implements CelerityReader {
	private Selector selector;
	private static final int HEAD_LENGTH = 4;
	private Channel<CelerityMessage> channel=ChannelFactory.messageChannel("READER");
	private SocketChannel sc;
	CelerityReaderImpl(SocketChannel sc) throws IOException {
		this.sc=sc;
	}

	@Override
	public void run() {
		try {
			while (true) {
				if (selector.select() <= 0) {
					continue;
				}
				Set<SelectionKey> set = selector.selectedKeys();
				Iterator<SelectionKey> it = set.iterator();
				while (it.hasNext()) {
					SelectionKey key = it.next();
					it.remove();
					if (key.isReadable()) {
						processRead(key);
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		System.out.println("read end");
	}

	@Override
	public void register(SocketChannel sc) throws IOException {
		sc.register(selector, SelectionKey.OP_READ, new CelerityBuffer());
		System.out.println("reader register");
	}

	@Override
	public void processRead(SelectionKey sk) throws IOException {
		CelerityBuffer cb = (CelerityBuffer) sk.attachment();
		if (!readHead(sk, cb)) {
			return;
		}
		if (!readBody(sk, cb)) {
			return;
		}
		try {
			Object object = parseObject(cb);
			System.out.println("reader: " + object);
			if (object instanceof CelerityMessage) {
				CelerityMessage message = (CelerityMessage) object;
				processMessage(message);
			}
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}

	private void processMessage(CelerityMessage message) {
		try {
			channel.offer(message);
		} catch (Exception e) {
			e.printStackTrace();
			return;
		}
	}

	private Object parseObject(CelerityBuffer buffer) throws IOException, ClassNotFoundException {
		ByteArrayInputStream bis = new ByteArrayInputStream(buffer.getInputBuffer().array());
		ObjectInputStream ois = new ObjectInputStream(bis);
		Object message = ois.readObject();
		buffer.getInputBuffer().clear();
		return message;
	}

	private boolean readBody(SelectionKey sk, CelerityBuffer buffer) throws IOException {
		SocketChannel sc = (SocketChannel) sk.channel();
		ByteBuffer bodyBuffer = buffer.getInputBuffer();
		sc.read(bodyBuffer);
		if (bodyBuffer.hasRemaining()) {
			return false;
		}
		return true;
	}

	private boolean readHead(SelectionKey sk, CelerityBuffer buffer) throws IOException {
		SocketChannel sc = (SocketChannel) sk.channel();
		ByteBuffer headBuffer = buffer.getHeadBuffer();
		sc.read(headBuffer);
		if (headBuffer.limit() == HEAD_LENGTH) {
			return true;
		}
		int length = 0;
		for (int i = 0; i < HEAD_LENGTH; i++) {
			length = length + ((int) headBuffer.get() & 0xFF) << (24 - i * 8);
		}
		return false;
	}

}
